package com.bkhech.bigdata.helloword;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;
import java.util.Arrays;
import java.util.Objects;

/**
 * 单词统计
 * <p>
 * 需求：读取hdfs上的 /data/wc_input/hello.txt 文件，计算文件中每个单词出现的总次数
 * hello.txt文件内容如下：
 * hello you
 * hello me
 * <p>
 * 最终需要的结果形式如下：
 * hello    2
 * me   1
 * you  1
 *
 * @author guowm
 * @date 2022/7/5
 */
@Slf4j
public class WordCountJob {

    /**
     * 创建自定义mapper类
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
        /**
         * 这个map函数就是可以接收k1,v1， 产生k2,v2
         *
         * @param k1 每一行的行首偏移量
         * @param v1 每一行内容
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable k1, Text v1, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
            //  k1代表的是每一行的行首偏移量，v1代表的是每一行内容
            //  对获取到的每一行数据进行切割，把单词切割出来
            final String[] words = v1.toString().split(" ");
            //  迭代切割出来的单词数据
            for (String word : words) {
                //  把迭代出来的单词封装成<k2,v2>的形式
                Text k2 = new Text(word);
                LongWritable v2 = new LongWritable(1L);
                log.info("k2: {}----v2: 1", word);
                //  把<k2,v2>写出到 hdfs
                context.write(k2, v2);
            }
        }
    }

    /**
     * 创建自定义的reducer类
     */
    public static class MySumReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        /**
         * 针对v2s的数据进行累加求和 并且最终把数据转化为k3,v3写出去
         *
         * @param k2
         * @param v2s
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text k2, Iterable<LongWritable> v2s, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
            // 创建一个sum变量，保存v2s的和
            long sum = 0L;
            for (LongWritable v2 : v2s) {
                sum += v2.get();
            }
            // 组装k3, v3
            Text k3 = k2;
            LongWritable v3 = new LongWritable(sum);
            // 把计算结果写出到 hdfs
            context.write(k3, v3);
        }
    }

    /**
     * hadoop集群上执行：
     * hadoop jar bigdata-0.0.1-SNAPSHOT-jar-with-dependencies.jar com.bkhech.bigdata.helloword.WordCountJob  /data/wc_input/hello.txt /tmp/word-count-job
     *
     * @param args
     */
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        //指定我们使用的文件系统类型:
        conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
            run(args, fileSystem);
            fileSystem.close();
        } catch (IOException e) {
            e.printStackTrace();
            if (Objects.nonNull(fileSystem)) {
                try {
                    fileSystem.close();
                } catch (IOException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }

    /**
     * 组装job = map + reduce
     *
     * @param args
     * @param fileSystem
     */
    public static void run(String[] args, FileSystem fileSystem) {
        try {
            // 可以指定job需要配置的参数
            final Configuration conf = fileSystem.getConf();

            String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
            if (otherArgs.length < 2) {
                System.err.println("Usage: wordcount <in> [<in>...] <out>");
                System.exit(2);
            }
            // 参数
            String[] inputDirs = Arrays.copyOf(otherArgs, otherArgs.length - 1);
            String outputDir = otherArgs[otherArgs.length - 1];
            int numOfReducers = 1;

            //  创建一个job
            final Job job = Job.getInstance(conf);

            //  注意：这一行必须设置，否则在集群中执行的是找不到WordCountJob这个类
            job.setJarByClass(WordCountJob.class);

            //  指定输入路径(可以是文件，也可以是目录)
            for (int i = 0; i < inputDirs.length; i++) {
                FileInputFormat.addInputPath(job, new Path(inputDirs[i]));
            }
            //  指定输出路径(只能指定一个不存在的目录)
            FileOutputFormat.setOutputPath(job, new Path(outputDir));

            //  指定map相关的配置
            job.setMapperClass(MyMapper.class);
            //  指定k2的类型
            job.setMapOutputKeyClass(Text.class);
            //  指定v2的类型
            job.setMapOutputValueClass(LongWritable.class);

            // 指定reduce相关的配置
            job.setReducerClass(MySumReducer.class);
            // 指定k3的类型
            job.setOutputKeyClass(Text.class);
            // 指定v3的类型
            job.setOutputValueClass(LongWritable.class);
            // 设置Reduce Task 数量，0 表示禁用 reduce
            job.setNumReduceTasks(numOfReducers);

            // job.submit(); // 异步提交
            // 同步提交 job
            boolean waitForCompletion = job.waitForCompletion(true);
            log.info("---execute result: {}---", waitForCompletion);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
